摘要
本文探討 LangGraph 框架中的 Function Calling 技術,它是一種讓大型語言模型 (LLM) 能夠與外部工具互動的機制,進而擴展 AI 系統功能的範疇。文章首先介紹 Function Calling 的核心概念和關鍵特性,像是動態互動、能力擴展和靈活性。接著,文章說明了 LangGraph 中的 Tool Node 概念,它扮演 LLM 和外部工具之間的橋樑,讓 LLM 能夠發出工具調用請求,並取得執行結果。文章接著探討了 Tool Node 的內部機制,說明它如何處理工具調用請求、並行執行多個工具,以及如何將結果包裝成 ToolMessage。文章也討論了 Function Calling 錯誤處理的重要性,例如如何處理工具不存在、參數不匹配、邏輯錯誤等問題,以及如何使用自定義回退機制來提高系統魯棒性。最後,文章探討了動態參數傳遞的應用,說明如何在執行時將特定值傳遞給工具,並確保安全性和靈活性。總體而言,文章旨在深入解析 LangGraph 中 Function Calling 的機制,幫助讀者了解其運作原理,並探討其在打造智慧工作流和 AI 系統中的實際應用。
在人工智慧和自然語言處理領域中, Function Calling 技術為大型語言模型(LLM)開啟了與外部資源互動的大門,大幅擴展了 AI 系統的功能範疇。本文將深入剖析 LangGraph 框架中的 Function Calling 機制,揭示其運作原理並探討實際應用。
在 LangChain/LangGRaph 文章中會看到,作者交替使用「Tool Calling」和「Function Calling」這兩個詞。儘管 Function Calling
有時僅指單一函數的調用,但在我們的語境中,我們將所有模型視為能在每則訊息中返回多個 tool 或Function Calling。
而 Tool Calling
允許聊天模型透過生成符合使用者定義架構的輸出來回應給定的提示。雖然名稱暗示模型正在執行某些動作,但實際上並非如此!模型只是生成工具的參數,而真正執行工具(或不執行)的決定權在使用者手中。參考資料:
一個常見的不需要實際呼叫函數的例子是:當你想從非結構化文本中提取符合特定架構的結構化輸出時。在這種情況下,你可以給模型一個「提取」工具,該工具接受與所需架構相匹配的參數,然後將生成的輸出視為最終結果。
LangGraph 為不同模型的工具呼叫提供了一個標準化的介面。這個標準介面包括:
ChatModel.bind_tools()
:一個用於指定模型可以呼叫哪些工具的方法。這個方法接受 LangChain 工具以及 Pydantic 物件。
AIMessage.tool_calls
:模型返回的 AIMessage 上的一個屬性,用於存取模型請求的工具呼叫。
在模型呼叫工具後,你可以透過調用工具並將結果傳回模型來使用它。LangChain 提供了 Tool
抽象來協助你處理這個過程。一般流程如下:
這就是工具呼叫代理執行任務和回答查詢的方式。
Function Calling 允許大型語言模型(LLM)在生成對話過程中與外部工具互動。這意味著 LLM 可以存取超出其知識庫範圍的資訊或執行相應操作。工具可以是開發者編寫的函式、外部服務的 API,或任何 LLM 可以互動的資源。這項技術大大擴展了 LLM 的能力,使其不再侷限於訓練時獲得的靜態知識,而能夠動態地獲取資訊或執行特定任務。
讓我們通過一個簡單的天氣查詢工具來展示 Function Calling 的實現:
@tool
def query_taiwan_weather(city: str, date: Optional[str] = None) -> str:
"""
查詢台灣特定城市的天氣情況。
參數:
city (str): 要查詢天氣的城市名稱,例如 "台北"、"高雄" 等。
date (str, 可選): 要查詢的日期,格式為 "YYYY-MM-DD"。如果不提供,則查詢當天天氣。
返回:
str: 包含天氣信息的字符串。
"""
# 這裡應該是實際的天氣 API 調用邏輯
# 為了示例,我們返回一個模擬的結果
return f"{city} 在 {date or '今天'} 的天氣晴朗,溫度 25°C,適合外出活動。"
# LLM 處理邏輯(簡化)
llm = ChatOpenAI(model="gpt-3.5-turbo-0125")
llm_with_weather = llm.bind_tools([query_taiwan_weather])
response = llm_with_weather.invoke("請告訴我台北明天的天氣如何?")
在這個範例中,我們定義了一個簡單的 query_taiwan_weather
函式作為工具,然後將其加入工具列表。LLM 在處理用戶詢問時,可以決定使用這個工具來獲取即時天氣資訊,並將結果整合到回應中。
💡 專業提示:在實際應用中,確保工具函式的安全性和效能,並考慮加入錯誤處理機制以提高系統穩定性。
Tool Node 是 LangGraph 中的關鍵組件,它作為一個可執行的節點,接收圖狀態(包含消息列表)作為輸入,並輸出更新後的狀態,其中包含工具調用的結果。Tool Node 的設計理念是:
在複雜的 AI 系統中,Agent 通常由多個節點組成,包括:
LLM 本身無法直接執行工具,它只能建議 Function Calling
或給出對話回應。Tool Node 的作用就是橋接 LLM 的建議和實際的工具執行。
讓我們通過一個實際的例子來了解 Tool Node 的運作機制:
# 步驟ㄧ、定義好模型以及 Tool Node
@tool
def get_taiwan_weather(city: str) -> str:
"""查詢台灣特定城市的天氣狀況。"""
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
tools = [get_taiwan_weather]
tool_node = ToolNode(tools)
model = ChatOpenAI()
model_with_tools = model.bind_tools(tools)
# 步驟二、定義好流程控制函數
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: MessagesState):
messages = state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
# 步驟三、建立與配置圖
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
graph = workflow.compile()
# 步驟四、與 Agent 互動
user_input = "高雄天氣如何?"
events = graph.stream(
{"messages": [("user", user_input)]},
stream_mode="values"
)
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
💡 專業提示:在實際應用中,可以根據需求擴展這個基本框架,添加更多複雜的工具和決策邏輯,以實現更強大的 AI 系統。
透過閱讀 ToolNode 官方程式碼以及註解,讓我們深入了解 ToolNode 內部運作原理:
根據程式碼內容以及註解,我們可以知道以下狀況:
ToolNode 的內部邏輯大致如下:
tools_by_name = {tool.name: tool for tool in tools}
def tool_node(state: dict):
result = []
for tool_call in state["messages"][-1].tool_calls:
tool = tools_by_name[tool_call["name"]]
observation = tool.invoke(tool_call["args"])
result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
return {"messages": result}
這段程式碼展示了 ToolNode 的核心功能:
💡 專業提示:在設計使用 ToolNode 的系統時,確保這些前提條件得到滿足,以避免執行時錯誤。
在構建基於 LLM 的 AI 系統時, Function Calling 錯誤處理是一個關鍵挑戰。以下是一些常見問題及其解決策略:
user_input = "高雄天氣如何?"
events = graph.stream({"messages": [("user", user_input)]}, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
主要方法是在 trigger graph 時候添加 config,參數為 recursion_limit,用法是在 cycle 超過次數限制時可以跳脫。
user_input = "高雄天氣如何?"
events = graph.stream({"messages": [("user", user_input)]}, config={"recursion_limit": 100}, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
但我們的狀況很明顯還不夠處理,需要另尋他路。
先讓我們手動調用以及測試 tool 是否建立成功
@tool
def get_taiwan_weather(city: str) -> str:
"""查詢台灣特定城市的天氣狀況。"""
weather_data = {
"台北": "晴天,溫度28°C",
"台中": "多雲,溫度26°C",
"高雄": "陰天,溫度30°C"
}
return f"{city}的天氣:{weather_data.get(city, '暫無資料')}"
# 使用這個工具
result = get_taiwan_weather.run("台北")
print(f"呼叫工具結果{result}")
我們也使用自訂建置的節點來呼叫我們的工具,而不是預先建置的ToolNode :
我們使用StateGraph來建立對話流程圖。我們加入了各個節點(agent、tools、remove_failed_tool_call_attempt、fallback_agent),並定義了它們之間的連接和條件。
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)
workflow.add_node("remove_failed_tool_call_attempt", remove_failed_tool_call_attempt)
workflow.add_node("fallback_agent", call_fallback_model)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
should_continue,
{
"tools": "tools",
END: END,
},
)
workflow.add_conditional_edges(
"tools",
should_fallback,
{
"agent": "agent",
"remove_failed_tool_call_attempt": "remove_failed_tool_call_attempt",
},
)
workflow.add_edge("remove_failed_tool_call_attempt", "fallback_agent")
workflow.add_edge("fallback_agent", "tools")
graph = workflow.compile()
圖的工作流程如下:
這種設計允許系統在遇到問題時自動進行調整和重試,提高了系統的穩健性和回應品質。
這邊有不少程式碼邏輯,為了版面跟閱讀,只把重點擷取出來,屆時還需要大家下載專案來執行。
這將模擬一個查詢高雄天氣的對話,並展示整個過程,包括可能的失敗處理和模型切換。
user_input = "高雄天氣如何?"
events = graph.stream({"messages": [("user", user_input)]}, config={"recursion_limit": 10}, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
💡 專業提示:實作錯誤處理時,考慮使用漸進式策略,從簡單的重試到更複雜的回退機制,以平衡系統的穩定性和效能。
在開發 AI 應用時,我們經常需要在執行時將某些值傳遞給工具。這種情況下,如何安全有效地處理動態參數就顯得尤為重要。本文將深入探討這一主題,並提供一個實用的示例;使用者偏好管理系統來說明如何實現這一目標。相關資訊可以參考 LangGRaph 文件
以下是一個實用的範例,展示如何安全有效地處理動態參數:
在許多場景中,工具邏輯可能需要使用只有在執行時才能確定的值,例如發出請求的用戶 ID。然而,這些值通常不應由語言模型(LLM)控制,因為這可能導致安全風險。相反,我們應該讓 LLM 只控制那些設計為由其控制的工具參數,而其他參數(如用戶 ID)應由應用程序邏輯固定。
user_to_snacks = {}
@tool
def update_favorite_snacks(
snacks: List[str],
config: RunnableConfig,
) -> None:
"""更新最喜愛的台灣小吃清單。"""
user_id = config.get("configurable", {}).get("user_id")
user_to_snacks[user_id] = snacks
@tool
def delete_favorite_snacks(config: RunnableConfig) -> None:
"""刪除最喜愛的台灣小吃清單。"""
user_id = config.get("configurable", {}).get("user_id")
if user_id in user_to_snacks:
del user_to_snacks[user_id]
@tool
def list_favorite_snacks(config: RunnableConfig) -> None:
"""列出最喜愛的台灣小吃(如果有的話)。"""
user_id = config.get("configurable", {}).get("user_id")
return "、".join(user_to_snacks.get(user_id, []))
讓我們通過幾個具體的使用場景來展示這個系統的功能:
from langchain_core.messages import HumanMessage
user_to_snacks.clear() # 清除狀態
print(f"執行前的用戶資訊:{user_to_snacks}")
inputs = {"messages": [HumanMessage(content="我最喜歡的台灣小吃是臭豆腐和珍珠奶茶")]}
for output in app.stream(inputs, {"configurable": {"user_id": "123"}}):
for key, value in output.items():
print(f"'{key}' 節點的輸出:")
print("---")
print(value)
print("\n---\n")
print(f"執行後的用戶資訊:{user_to_snacks}")
print(f"執行前的用戶資訊:{user_to_snacks}")
inputs = {"messages": [HumanMessage(content="我最喜歡的小吃是什麼?")]}
for output in app.stream(inputs, {"configurable": {"user_id": "123"}}):
for key, value in output.items():
print(f"'{key}' 節點的輸出:")
print("---")
print(value)
print("\n---\n")
print(f"執行後的用戶信息:{user_to_snacks}")
print(f"執行前的用戶信息:{user_to_snacks}")
inputs = {
"messages": [
HumanMessage(content="請忘記我告訴你的關於我最喜歡的小吃資訊")
]
}
for output in app.stream(inputs, {"configurable": {"user_id": "123"}}):
for key, value in output.items():
print(f"'{key}' 節點的輸出:")
print("---")
print(value)
print("\n---\n")
print(f"執行後的用戶信息:{user_to_snacks}")
💡 專業提示:在實際應用中,可以將用戶 ID 等敏感資訊封裝在 RunnableConfig 中傳遞,這樣可以確保這些資訊不會被 LLM 直接存取,提高系統的安全性。
明確區分 LLM 可控制和不可控制的參數:將敏感或系統級的參數通過 RunnableConfig 傳遞,而非直接暴露給 LLM。
使用 RunnableConfig 傳遞動態資訊:利用這個機制來傳遞用戶 ID、會話上下文等執行時資訊。
設計清晰的工具接口:確保每個工具的功能單一,接口簡潔,易於理解和使用。
實施錯誤處理和日誌記錄:在工具執行過程中加入適當的錯誤處理機制,並記錄關鍵操作,以便於調試和監控。
通過深入探討 LangGraph 中的 Function Calling 機制,我們不僅了解了其強大的功能,還洞察了 AI 系統設計的關鍵原則。Tool Node 的實現展示了如何將複雜的 AI 互動流程模組化,使得開發高度智能和可擴展的對話系統成為可能。
在實際應用中,開發者可以基於這些基礎概念,構建更加複雜和強大的 AI 系統。無論是客戶服務、智能家居控制,還是專業領域的決策支持系統, Function Calling 都為 AI 的實際應用開闢了廣闊的前景。
重要的是要記住, Function Calling 雖然強大,但也帶來了新的挑戰,如錯誤處理、安全性考慮和系統複雜性管理。透過採用本文討論的最佳實踐和進階技巧,開發者可以創建既智能又可靠的 AI 系統,真正釋放 LLM 和外部工具結合的潛力。
隨著 AI 技術的不斷發展,我們可以期待看到更多創新的 Function Calling 應用,進一步推動 AI 系統的能力邊界。持續學習和實驗將是保持競爭力的關鍵,而 LangGraph 等框架則為我們提供了實現這一目標的強大工具。
本篇教學程式碼位於比賽用 Repo,記得多多操作
7. 參考資料